﻿/*
   Copyright 2012 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Flower.Client;
using Flower.Services.Data;
using Flower.Workflow;
using Flower.Processing.Interruptions;
using Flower.Directory.Util;
using Flower.Services;

namespace Flower.Processing.Activities
{
    [SaveAfter]
    [Transactional]
    [Retry]
    public class Dequeue<TMsg> : WaitBase
    {
        private interface IDequeue
        {
            Blob Dequeue();
        }

        private class SetDequeue : IDequeue
        {
            private readonly Processor _processor;
            private readonly string _subset;

            public SetDequeue(Processor processor, string subset)
            {
                _processor = processor;
                _subset = subset;
            }

            public Blob Dequeue()
            {
                _processor.Log.Trace(() => string.Format("Executing __dequeueMessage({0}, {1}).", _subset, _processor.Pid));

                Message message =
                    ScriptHelper.Dequeue
                    (
                        _processor.Client,
                        _subset,
                        _processor.Pid
                    );

                return (message != null) ? message.Data : null;
            }
        }

        private class QueueDequeue : IDequeue
        {
            private readonly Processor _processor;
            private readonly IQueueService _queue;
            private readonly string _recipient;
            private readonly string _servicePath;

            public QueueDequeue(Processor processor, IQueueService queue, string recipient, string servicePath)
            {
                _processor = processor;
                _queue = queue;
                _recipient = recipient;
                _servicePath = servicePath;
            }

            public Blob Dequeue()
            {
                _processor.Log.Trace(() => string.Format("Executing IQueueService.Dequeue({0}) at {1}.", _recipient, _servicePath));

                return _queue.Dequeue(_recipient, _processor.Pid);
            }
        }

        internal Func<IContext, string> Subset { get; set; }
        internal Action<IContext, TMsg> Message { get; set; }

        internal Dequeue() { }

        public override bool IsBlocking
        {
            get { return true; }
        }

        private static IDequeue ResolvePath(Processor processor, string path)
        {
            if (path == null)
            {
                var servicePath = string.Concat("/Processors/", processor.ProcessorName, "/queue");
                var queue = processor.Client.GetService<IQueueService>(servicePath);

                return new QueueDequeue(processor, queue, string.Concat(processor.Pid, "/", typeof(TMsg).FullName), processor.Pid);
            }
            else if (DirectoryUtil.IsId(path))
            {
                IQueueService queue = processor.Client.GetProcessQueue(path);

                return new QueueDequeue(processor, queue, string.Concat(path, "/", typeof(TMsg).FullName), path);
            }
            else if (path.StartsWith("/Services"))
            {
                string servicePath, recipient;
                DirectoryUtil.CutPathTailSegment(path, out servicePath, out recipient);

                var queue = processor.Client.GetService<IQueueService>(servicePath);

                return new QueueDequeue(processor, queue, recipient, servicePath);
            }
            else
            {
                string subset = DirectoryUtil.LocalSubsetToGlobal(processor.Pid, path);

                return new SetDequeue(processor, subset);
            }
        }

        public override Interruption Do(Processor processor)
        {
            string subset = (Subset != null) ? Subset(processor) : null;

            IDequeue deq = ResolvePath(processor, subset);

            Blob message = deq.Dequeue();

            if (message == null)
            {
                if (Interrupt)
                {
                    return Waiting.Instance;
                }
                else
                {
                    if (deq is SetDequeue)
                    {
                        processor.State.__WaitLinks.Add
                        (
                            string.Concat(DirectoryUtil.LocalSubsetToGlobal(processor.Pid, subset), "/GetWaiters/", processor.Pid)
                        );
                    }

                    ++processor.State.__ActivityPointer;
                    return null;
                }
            }
            else
            {
                if (Message != null)
                {
                    Message(processor, (message != null) ? message.To<TMsg>() : default(TMsg));
                }

                processor.State.__ActivityPointer = Position;
                return null;
            }
        }
    }
}
